#include "conn_event.h"
#include "process.h"
#include "transfer.h"

void connev_set_address(struct connev_t* connev)
{
	assert(connev && connev->ev);
	int32_t sock_fd = event_get_fd(connev->ev);
	
	/* address */
	struct sockaddr_in addr;
	socklen_t len;
	memset(connev->peer_addr, 0, sizeof(connev->peer_addr));
	memset(connev->self_addr, 0, sizeof(connev->self_addr));

	/* peer */
	len = sizeof(addr);
	getpeername(sock_fd, (struct sockaddr*)&addr, &len);
	inet_ntop(AF_INET, &(addr.sin_addr), connev->peer_addr, INET_ADDRSTRLEN);
	snprintf(connev->peer_addr + strlen(connev->peer_addr), 
		sizeof(connev->peer_addr) - strlen(connev->peer_addr),
		":%d", (int32_t)htons(addr.sin_port));

	/* self */
	len = sizeof(addr);
	getsockname(sock_fd, (struct sockaddr*)&addr, &len);
	inet_ntop(AF_INET, &(addr.sin_addr), connev->self_addr, INET_ADDRSTRLEN);
	snprintf(connev->self_addr + strlen(connev->self_addr), 
		sizeof(connev->self_addr) - strlen(connev->self_addr),
		":%d", (int32_t)htons(addr.sin_port));
}

/* create a new connection event */
struct connev_t* connev_init(evutil_socket_t sock_fd, int32_t connbuffer_size)
{
	/* event handler */
	struct connev_t* connev = (struct connev_t*)MALLOC(sizeof(struct connev_t));
	connev->ev = event_new(gCtx->reactor, sock_fd, EV_READ | EV_PERSIST,
		connev_callback, connev);

	/* waiting list */
	INIT_LIST_HEAD(&connev->waiting);

	/* status */
	connev->status = connev_status_wait_syn;

	/* address */
	connev->address = gCtx->ev_address ++;

	/* buffer */
	connev->in_buffer = connbuffer_init(connbuffer_size, connbuffer_size / 3);
	connev->out_buffer = connbuffer_init(connbuffer_size, connbuffer_size / 3);
	assert(connev && connev->ev && connev->in_buffer && connev->out_buffer);

	/* address */
	connev_set_address(connev);

	/* non-block */
	evutil_make_socket_nonblocking(sock_fd);
	
	return connev;
}

int32_t connev_add(struct connev_t* connev)
{
	assert(connev);
	struct timeval tv;
	tv.tv_sec = gCtx->cfg->kick_timeout;
	tv.tv_usec = 0;
	return event_add(connev->ev, &tv);
}

int32_t connev_del(struct connev_t* connev)
{
	assert(connev);
	return event_del(connev->ev);
}

/* update register events */
void connev_update_event(struct connev_t* connev, short events)
{
	connev_del(connev);
	event_assign(connev->ev, gCtx->reactor, event_get_fd(connev->ev), 
		events, connev_callback, connev);
	connev_add(connev);
}

/* reset connection event, ensure to call it after remove from reactor event base */
void connev_reset(struct connev_t* connev)
{
	assert(connev);
	connbuffer_reset(connev->in_buffer);
	connbuffer_reset(connev->out_buffer);
	event_free(connev->ev);
	connev->ev = 0;
	connev->address = -1;
	connev->status = connev_status_none;
	INIT_LIST_HEAD(&connev->waiting);
	memset(connev->peer_addr, 0, sizeof(connev->peer_addr));
	memset(connev->self_addr, 0, sizeof(connev->peer_addr));
}

/* re-init reset connection event */
void connev_reinit(struct connev_t* connev, evutil_socket_t sock_fd)
{
	assert(connev);
	assert(0 == connev->ev);
	connev->ev = event_new(gCtx->reactor, sock_fd, EV_READ  | EV_PERSIST,
		connev_callback, connev);
	assert(connev->ev);

	/* waiting list */
	INIT_LIST_HEAD(&connev->waiting);

	/* status */
	connev->status = connev_status_wait_syn;

	/* address */
	connev->address = gCtx->ev_address ++;

	/* address */
	connev_set_address(connev);
	
	/* non-block */
	evutil_make_socket_nonblocking(sock_fd);
}

/* destruct connection-event, remove from reactor, recycle resource */
void connev_destruct(struct connev_t* connev, int32_t notify_type)
{
	assert(connev && connev->ev);

	/* 
	* if need to notify to a client, send notification first,
	* do destructtion after sending complete
	*/
	
	if(connev_notify_client_version_fail & notify_type)
	{
		transfer_conn2c(ID_CONN2C_NTF_DISCONN, KICK_VERSION_FAIL, connev);
		connev_update_event(connev, EV_WRITE);
		connev->notify_type = (~connev_notify_client_version_fail & notify_type);
		return;
	}

	if(connev_notify_client_server_disconn & notify_type)
	{
		transfer_conn2c(ID_CONN2C_NTF_DISCONN, KICK_SERVER_DISCONNECT, connev);
		connev->notify_type = (~connev_notify_client_server_disconn & notify_type);
		connev_update_event(connev, EV_WRITE);
		return;
	}
	if(connev_notify_client_timeout_fail & notify_type)
	{
		transfer_conn2c(ID_CONN2C_NTF_DISCONN, KICK_SERVER_TIMEOUT, connev);
		connev->notify_type = (~connev_notify_client_timeout_fail & notify_type);
		connev_update_event(connev, EV_WRITE);
		return;
	}

	LOG_DEBUG("destruct connection from %s", connev_peer_addr(connev));

	/* update statics */
	int32_t pop_flag = 0;

	/* connected destruct */
	if(connev->status == connev_status_connected)
	{	
		gCtx->connected_conn_count --;

		/* check if waiting connections */
		if(gCtx->waiting_conn_count > 0)
		{
			pop_flag = 1;
		}
	}

	/* in-queue destruct */
	else if(connev->status == connev_status_in_queue)
	{
		gCtx->waiting_conn_count --;

		/* remove list */
		list_del(&connev->waiting);
	}

	/* remove from reactor */
	connev_del(connev);

	/* notify to client or server */
	if(connev_notify_server & notify_type)
	{
		transfer_conn2s(ID_CONN2S_NTF_DISCONN, 
			connev->address, gCtx->lsi, gCtx->cfg->lsi_server_ip);
	}

	/* close fd */
	close(event_get_fd(connev->ev));

	/* status */
	connev->status = connev_status_none;
	
	/* remove from hash list */
	hash_remove(gCtx->connev_table, connev);

	/* release */
	connev_pool_free(gCtx->connev_pool, connev);

	/* pop waiting queue */
	if(pop_flag)
		connev_proc_pop_queue();
	
	return;
}

/* event callback */
void connev_callback(evutil_socket_t fd, short event, void* args)
{
	struct connev_t* connev = (struct connev_t*)(args);
	assert(event_get_fd(connev->ev) == fd);

	/* timeout event */
	if(event & EV_TIMEOUT)
		if(connev_timeout_cb(connev) < 0)
			return;
		
	/* read event */
	if(event & EV_READ)
		if(connev_read_cb(connev) < 0)
			return;

	/* write event */
	if(event & EV_WRITE)
		if(connev_write_cb(connev) < 0)
			return;
}

/*
* return < 0, connev destructed
*/
int32_t connev_timeout_cb(struct connev_t* connev)
{
	assert(connev && connev->ev);
	LOG_WARN("connev from %s timeout, destruction", connev_peer_addr(connev));
	connev_destruct(connev, connev_notify_server |connev_notify_client_timeout_fail);
	return -1;
}

/*
* return < 0, connev destructed
*/
int32_t connev_read_cb(struct connev_t* connev)
{
	/* set wait flag */
	gCtx->wait_flag = 0;
	assert(connev && connev->ev);
	
	/*	
	*	if in-buffer is full, maybe we'll miss peer's disconnect signal, 
	*	so heartbeat is essential in this case
	*/
	int32_t write_len = connbuffer_write_len(connev->in_buffer);
	if(write_len <= 0)
	{
		LOG_DEBUG("connev from %s write buffer full, do read first", connev_peer_addr(connev));
		return connev_process(connev);
	}
	
	/*	try recv data from socket */
	int32_t sock_fd = event_get_fd(connev->ev);
	int32_t ret = recv(sock_fd, connbuffer_write_buffer(connev->in_buffer), write_len, 0);
	if(ret < 0)
	{
		/* blocked or get signal*/
		if (EAGAIN == errno || EWOULDBLOCK == errno || EINTR == errno)
		{
			LOG_DEBUG("connev from %s, read error: %s, ignore", 
				connev_peer_addr(connev), strerror(errno));
		}
		/* connection reset */
		else
		{
			goto READ_FAIL;
		}
	}
	/* peer close connection */
	else if(0 == ret)
	{
		goto READ_FAIL;
	}
	/* read bytes */
	else
	{
		/* write ptr */
		connbuffer_write_nocopy(connev->in_buffer, ret);

		/* update events, change timeout */
		connev_update_event(connev, event_get_events(connev->ev));

		/* process read data */
		return connev_process(connev);
	}

	return 0;

READ_FAIL:
	LOG_ERROR("connev from %s read errno=%d, %s, destruct connection", 
				connev_peer_addr(connev), errno, strerror(errno));
	if(connev->status == connev_status_connected)
		connev_destruct(connev, connev_notify_server);
	else 
		connev_destruct(connev, connev_notify_none);
	return -1;
}

/*
* return < 0, connev destructed
*/
int32_t connev_write_cb(struct connev_t* connev)
{
	/* set wait flag */
	gCtx->wait_flag = 0;
	assert(connev && connev->ev);

	/* has any data need to write to socket */
	int32_t data_len = connbuffer_read_len(connev->out_buffer);
	short events = event_get_events(connev->ev);
	
	if(data_len <= 0)
	{	
		LOG_WARN("connev from %s no buffer to write, ignore", connev_peer_addr(connev));

		/* if only EV_WRITE, means events need to destruct */
		if(events == EV_WRITE)
		{
			connev_destruct(connev, connev->notify_type);
			return 0;
		}
		else
		{
			events &= (~EV_WRITE);
			connev_update_event(connev, events);
			return 0;	
		}
	}

	/* try write to socket */
	int32_t nwrite = send(event_get_fd(connev->ev), 
		connbuffer_read_buffer(connev->out_buffer), data_len, 0);

	/* write fail */
	if(nwrite <= 0)
	{
		if(EAGAIN == errno || EWOULDBLOCK == errno || EINTR == errno)
		{
			LOG_DEBUG("connev from %s, write error: %s, ignore", 
				connev_peer_addr(connev), strerror(errno));
			return 0;
		}
		else
		{
			goto WRITE_FAIL;
		}
	}

	/* read ptr */
	connbuffer_read_nocopy(connev->out_buffer, nwrite);

	/* if write complete, remove write event */
	if(nwrite == data_len)
	{
		/* if only EV_WRITE, means events need to destruct */
		if(events == EV_WRITE)
		{
			connev_destruct(connev, connev->notify_type);
			return 0;
		}
		else
		{
			events &= (~EV_WRITE);
			connev_update_event(connev, events);	
		}
	}

	return 0;

WRITE_FAIL:
	LOG_DEBUG("connev from %s, write error=%d: %s, destruct connection", 
				connev_peer_addr(connev), errno, strerror(errno));
	connev_destruct(connev, connev_notify_server);
	return -1;
}

/* 
*  write data to connection event buffer 
*  return = 0 success
*  return < 0 fail
*/
int32_t connev_write(struct connev_t* connev, const char* buffer, size_t buffer_len)
{
	assert(connev && connev->ev);

	/* check enough buffer to write */
	int32_t write_len = connbuffer_write_len(connev->out_buffer);
	if(buffer_len > write_len)
	{
		LOG_ERROR("out buffer write len=%d, write %d bytes fail", write_len, buffer_len);
		return -1;
	}

	/* write data to buffer */
	int32_t nwrite = connbuffer_write(connev->out_buffer, buffer, buffer_len);
	assert(nwrite == buffer_len);

	/* add write event */
	short events = event_get_events(connev->ev);
	events |= EV_WRITE;
	connev_update_event(connev, events);
	return 0;
}

int32_t connev_hash(const void* ev)
{
	const struct connev_t* connev = (const struct connev_t*)ev;
	return connev->address;
}

int32_t connev_cmp(const void* ev1, const void* ev2)
{
	const struct connev_t* connev1 = (const struct connev_t*)ev1;
	const struct connev_t* connev2 = (const struct connev_t*)ev2;
	return connev1->address - connev2->address;
}

/* release connection event resource */
void connev_release(struct connev_t* connev)
{
	assert(connev);
	if(connev->ev)
		event_free(connev->ev);
	connbuffer_release(connev->in_buffer);
	connbuffer_release(connev->out_buffer);
	FREE(connev);
	connev = 0;
}

const char* connev_peer_addr(struct connev_t* connev)
{
	assert(connev);
	return connev->peer_addr;
}

const char* connev_self_addr(struct connev_t* connev)
{
	assert(connev);
	return connev->self_addr;
}

typedef struct connev_pool_t
{
	/* connection event buffer size */
	int32_t connbuffer_size;

	/* threshold to release connev */
	int32_t pool_threshold;

	/* free connev list */
	int32_t free_connev_count;
	struct connev_t** free_connev_list;
}connev_pool_t;

struct connev_pool_t* connev_pool_init(int32_t connbuffer_size, int32_t threshold_hint)
{
	assert(threshold_hint > 0);
	struct connev_pool_t* connev_pool = (struct connev_pool_t*)MALLOC(sizeof(struct connev_pool_t));
	connev_pool->connbuffer_size = connbuffer_size;
	connev_pool->pool_threshold = threshold_hint;
	connev_pool->free_connev_count = 0;
	connev_pool->free_connev_list = (struct connev_t**)CALLOC(connev_pool->pool_threshold,
			sizeof(struct connev_t*));
	LOG_DEBUG("connection event pool inited with threshold=%d", threshold_hint);
	return connev_pool;
}

void connev_pool_release(struct connev_pool_t* pool)
{
	if(pool)
	{
		int32_t i;
		for(i=0; i<pool->free_connev_count; i++)
			connev_release(pool->free_connev_list[i]);
		FREE(pool->free_connev_list);
		FREE(pool);
	}
	LOG_DEBUG("connection event pool release");
}

/* allocate a connection-event from pool and init */
struct connev_t* connev_pool_alloc_init(struct connev_pool_t* pool, evutil_socket_t sock_fd)
{
	assert(pool);
	
	/* alloc from free list */
	struct connev_t* connev = 0;
	if(pool->free_connev_count > 0)
	{
		connev = pool->free_connev_list[pool->free_connev_count - 1];
		pool->free_connev_count --;
		connev_reinit(connev, sock_fd);
	}
	/* new one */
	else
	{
		connev = connev_init(sock_fd, pool->connbuffer_size);
	}
	
	return connev;
}

void connev_pool_free(struct connev_pool_t* pool, struct connev_t* connev)
{
	assert(pool && connev);
	
	/* release direct */
	if(pool->free_connev_count >= pool->pool_threshold)
	{
		connev_release(connev);
		connev = 0;
	}
	/* add to free list */
	else
	{
		connev_reset(connev);
		pool->free_connev_list[pool->free_connev_count] = connev;
		pool->free_connev_count ++;
	}
}

